package com.zeekr.rocketmq.producer;

import cn.hutool.json.JSONUtil;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.filter.impl.Type;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.time.LocalTime;
import java.util.ArrayList;
import java.util.List;

/**
 * @author yangyuanping
 */

@Component
public class MyProducer {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 同步消息
     *
     * @param topic
     * @param mess
     */
    public void sendSyncMessage(String topic, String mess) {
        for (int i = 0; i < 1; i++) {
            SendResult sendResult = rocketMQTemplate.syncSend(topic, i + "_" + mess);
            System.out.println("消息发送结果:" + JSONUtil.toJsonStr(sendResult));
        }
    }

    /**
     * 异步消息
     *
     * @param topic
     * @param mess
     */
    public void sendAsyncMessage(String topic, String mess) {
        for (int i = 0; i < 5; i++) {
            Message<String> message = MessageBuilder.withPayload(i + "_" + mess).build();
            rocketMQTemplate.asyncSend(topic, message, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    System.out.println("发送成功");
                }

                @Override
                public void onException(Throwable e) {
                    System.out.println("方式异常：" + e.getMessage());
                }
            });
        }
    }
    /**
     * 顺序消息
     *
     * @param topic
     * @param mess
     */
    public void sendOrderMessage(String topic, String mess) {
        for (int i = 0; i < 10; i++) {
            Message<String> message = MessageBuilder.withPayload(i + "_" + mess).build();
            SendResult sendResult = rocketMQTemplate.syncSendOrderly(topic, message, "one");
            System.out.println("顺序消息发送结果:"+sendResult);
        }
    }
    /**
     * 延时消息
     *
     * @param topic
     * @param mess
     */
    public void sendDelayedMessage(String topic, String mess) {
        for (int i = 0; i < 1; i++) {
            Message<String> message = MessageBuilder.withPayload(i + "_" + mess).build();
            SendResult sendResult = rocketMQTemplate.syncSend(topic, message, 5000, 2);
            System.out.println("延时消息发送结果:" + LocalTime.now() + "-" + sendResult);
        }
    }

    /**
     * 批量消息
     *
     * @param topic
     * @param mess
     */
    public void sendBatchMessage(String topic, String mess) {
        List<Message> messages = new ArrayList<>();
        for (int i = 0; i < 5; i++) {
            Message<String> message = MessageBuilder.withPayload(i + "_" +mess).build();
            messages.add(message);
        }
        SendResult sendResult = rocketMQTemplate.syncSend(topic, messages);
        System.out.println("消息批量发送结果:" + JSONUtil.toJsonStr(sendResult));
    }
}
